{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 数据库交互式分析模板\n",
    "\n",
    "本笔记本提供了与 alphahome 数据库进行交互式分析的完整示例，包括：\n",
    "- 数据库连接和配置\n",
    "- 数据查询和探索\n",
    "- 数据质量检查\n",
    "- 基础统计分析\n",
    "- 数据可视化\n",
    "\n",
    "## 环境设置"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 导入必要的库\n",
    "import sys\n",
    "import os\n",
    "import asyncio\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import seaborn as sns\n",
    "from datetime import datetime, timedelta\n",
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "# 设置中文字体\n",
    "plt.rcParams['font.sans-serif'] = ['SimHei', 'Microsoft YaHei']\n",
    "plt.rcParams['axes.unicode_minus'] = False\n",
    "\n",
    "# 添加项目根目录到路径\n",
    "project_root = os.path.abspath(os.path.join(os.getcwd(), '../../../..'))\n",
    "if project_root not in sys.path:\n",
    "    sys.path.insert(0, project_root)\n",
    "\n",
    "print(f\"项目根目录: {project_root}\")\n",
    "print(f\"当前工作目录: {os.getcwd()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据库连接设置"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 导入 alphahome 组件\n",
    "from alphahome.common.config_manager import ConfigManager\n",
    "from alphahome.common.db_manager import DBManager\n",
    "\n",
    "# 初始化配置管理器\n",
    "config_manager = ConfigManager()\n",
    "config = config_manager.load_config()\n",
    "\n",
    "print(\"配置加载成功\")\n",
    "print(f\"数据库URL: {config.get('database', {}).get('url', 'Not configured')}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 创建数据库连接\n",
    "async def create_db_connection():\n",
    "    \"\"\"创建异步数据库连接\"\"\"\n",
    "    db_manager = DBManager()\n",
    "    await db_manager.initialize()\n",
    "    return db_manager\n",
    "\n",
    "# 在 Jupyter 中运行异步代码\n",
    "try:\n",
    "    # 检查是否已有事件循环\n",
    "    loop = asyncio.get_event_loop()\n",
    "    if loop.is_running():\n",
    "        # 在已运行的事件循环中创建任务\n",
    "        import nest_asyncio\n",
    "        nest_asyncio.apply()\n",
    "        db = await create_db_connection()\n",
    "    else:\n",
    "        db = await create_db_connection()\n",
    "except RuntimeError:\n",
    "    # 创建新的事件循环\n",
    "    db = asyncio.run(create_db_connection())\n",
    "\n",
    "print(\"数据库连接创建成功\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据库结构探索"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 查看所有表\n",
    "async def get_all_tables():\n",
    "    \"\"\"获取数据库中所有表的信息\"\"\"\n",
    "    query = \"\"\"\n",
    "    SELECT \n",
    "        schemaname,\n",
    "        tablename,\n",
    "        tableowner\n",
    "    FROM pg_tables \n",
    "    WHERE schemaname NOT IN ('information_schema', 'pg_catalog')\n",
    "    ORDER BY schemaname, tablename;\n",
    "    \"\"\"\n",
    "    return await db.fetch_all(query)\n",
    "\n",
    "tables = await get_all_tables()\n",
    "tables_df = pd.DataFrame(tables)\n",
    "\n",
    "print(f\"数据库中共有 {len(tables_df)} 个表:\")\n",
    "print(tables_df.to_string(index=False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 查看表的详细信息\n",
    "async def get_table_info(table_name, schema='public'):\n",
    "    \"\"\"获取指定表的列信息\"\"\"\n",
    "    query = \"\"\"\n",
    "    SELECT \n",
    "        column_name,\n",
    "        data_type,\n",
    "        is_nullable,\n",
    "        column_default\n",
    "    FROM information_schema.columns \n",
    "    WHERE table_schema = $1 AND table_name = $2\n",
    "    ORDER BY ordinal_position;\n",
    "    \"\"\"\n",
    "    return await db.fetch_all(query, schema, table_name)\n",
    "\n",
    "# 示例：查看股票基础信息表结构（如果存在）\n",
    "sample_tables = ['tushare_stock_basic', 'tushare_stock_daily', 'stock_basic']\n",
    "\n",
    "for table_name in sample_tables:\n",
    "    try:\n",
    "        table_exists = await db.table_exists(table_name)\n",
    "        if table_exists:\n",
    "            print(f\"\\n=== {table_name} 表结构 ===\")\n",
    "            columns = await get_table_info(table_name)\n",
    "            columns_df = pd.DataFrame(columns)\n",
    "            print(columns_df.to_string(index=False))\n",
    "            break\n",
    "    except Exception as e:\n",
    "        print(f\"检查表 {table_name} 时出错: {e}\")\n",
    "        continue\n",
    "else:\n",
    "    print(\"未找到常见的股票数据表，请检查数据库中的表名\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据查询和探索"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 查询示例数据\n",
    "async def query_sample_data(table_name, limit=10):\n",
    "    \"\"\"查询表的示例数据\"\"\"\n",
    "    query = f\"SELECT * FROM {table_name} LIMIT $1\"\n",
    "    return await db.fetch_all(query, limit)\n",
    "\n",
    "# 获取数据表行数统计\n",
    "async def get_table_stats():\n",
    "    \"\"\"获取各表的行数统计\"\"\"\n",
    "    stats = []\n",
    "    for table in tables:\n",
    "        table_name = table['tablename']\n",
    "        schema_name = table['schemaname']\n",
    "        full_name = f\"{schema_name}.{table_name}\" if schema_name != 'public' else table_name\n",
    "        \n",
    "        try:\n",
    "            count_query = f\"SELECT COUNT(*) as count FROM {full_name}\"\n",
    "            result = await db.fetch_one(count_query)\n",
    "            stats.append({\n",
    "                'schema': schema_name,\n",
    "                'table': table_name,\n",
    "                'row_count': result['count']\n",
    "            })\n",
    "        except Exception as e:\n",
    "            print(f\"查询表 {full_name} 行数时出错: {e}\")\n",
    "            stats.append({\n",
    "                'schema': schema_name,\n",
    "                'table': table_name,\n",
    "                'row_count': 'Error'\n",
    "            })\n",
    "    \n",
    "    return stats\n",
    "\n",
    "# 执行统计查询\n",
    "table_stats = await get_table_stats()\n",
    "stats_df = pd.DataFrame(table_stats)\n",
    "\n",
    "print(\"数据表行数统计:\")\n",
    "print(stats_df.to_string(index=False))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 选择一个有数据的表进行详细分析\n",
    "# 找到行数最多的表\n",
    "numeric_stats = stats_df[stats_df['row_count'] != 'Error'].copy()\n",
    "if not numeric_stats.empty:\n",
    "    numeric_stats['row_count'] = pd.to_numeric(numeric_stats['row_count'])\n",
    "    largest_table = numeric_stats.loc[numeric_stats['row_count'].idxmax()]\n",
    "    \n",
    "    target_table = largest_table['table']\n",
    "    target_schema = largest_table['schema']\n",
    "    full_table_name = f\"{target_schema}.{target_table}\" if target_schema != 'public' else target_table\n",
    "    \n",
    "    print(f\"选择表 '{full_table_name}' 进行详细分析 (行数: {largest_table['row_count']:,})\")\n",
    "    \n",
    "    # 查看示例数据\n",
    "    sample_data = await query_sample_data(full_table_name, 5)\n",
    "    sample_df = pd.DataFrame(sample_data)\n",
    "    \n",
    "    print(f\"\\n{target_table} 示例数据:\")\n",
    "    print(sample_df.head())\n",
    "    \n",
    "    # 获取表结构\n",
    "    table_columns = await get_table_info(target_table, target_schema)\n",
    "    columns_df = pd.DataFrame(table_columns)\n",
    "    \n",
    "    print(f\"\\n{target_table} 表结构:\")\n",
    "    print(columns_df.to_string(index=False))\n",
    "    \n",
    "else:\n",
    "    print(\"未找到可分析的数据表\")\n",
    "    target_table = None"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据质量检查"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "if target_table:\n",
    "    # 数据质量检查\n",
    "    async def data_quality_check(table_name):\n",
    "        \"\"\"执行数据质量检查\"\"\"\n",
    "        quality_report = {}\n",
    "        \n",
    "        # 总行数\n",
    "        count_result = await db.fetch_one(f\"SELECT COUNT(*) as total_rows FROM {table_name}\")\n",
    "        quality_report['total_rows'] = count_result['total_rows']\n",
    "        \n",
    "        # 获取所有列名\n",
    "        columns_query = \"\"\"\n",
    "        SELECT column_name, data_type \n",
    "        FROM information_schema.columns \n",
    "        WHERE table_name = $1 AND table_schema = 'public'\n",
    "        ORDER BY ordinal_position\n",
    "        \"\"\"\n",
    "        columns = await db.fetch_all(columns_query, table_name.split('.')[-1])\n",
    "        \n",
    "        # 检查每列的空值情况\n",
    "        null_checks = []\n",
    "        for col in columns:\n",
    "            col_name = col['column_name']\n",
    "            null_query = f\"\"\"\n",
    "            SELECT \n",
    "                COUNT(*) as total_count,\n",
    "                COUNT({col_name}) as non_null_count,\n",
    "                COUNT(*) - COUNT({col_name}) as null_count\n",
    "            FROM {table_name}\n",
    "            \"\"\"\n",
    "            try:\n",
    "                result = await db.fetch_one(null_query)\n",
    "                null_checks.append({\n",
    "                    'column': col_name,\n",
    "                    'data_type': col['data_type'],\n",
    "                    'total_count': result['total_count'],\n",
    "                    'non_null_count': result['non_null_count'],\n",
    "                    'null_count': result['null_count'],\n",
    "                    'null_percentage': (result['null_count'] / result['total_count'] * 100) if result['total_count'] > 0 else 0\n",
    "                })\n",
    "            except Exception as e:\n",
    "                print(f\"检查列 {col_name} 时出错: {e}\")\n",
    "        \n",
    "        quality_report['column_quality'] = null_checks\n",
    "        return quality_report\n",
    "    \n",
    "    # 执行数据质量检查\n",
    "    quality_report = await data_quality_check(full_table_name)\n",
    "    \n",
    "    print(f\"数据质量报告 - {target_table}\")\n",
    "    print(f\"总行数: {quality_report['total_rows']:,}\")\n",
    "    print(\"\\n列质量统计:\")\n",
    "    \n",
    "    quality_df = pd.DataFrame(quality_report['column_quality'])\n",
    "    print(quality_df[['column', 'data_type', 'null_count', 'null_percentage']].to_string(index=False))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 自定义查询区域\n",
    "\n",
    "在这里你可以执行自定义的SQL查询来探索数据"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 自定义查询示例\n",
    "async def execute_custom_query(query, params=None):\n",
    "    \"\"\"执行自定义查询\"\"\"\n",
    "    try:\n",
    "        if params:\n",
    "            result = await db.fetch_all(query, *params)\n",
    "        else:\n",
    "            result = await db.fetch_all(query)\n",
    "        return pd.DataFrame(result)\n",
    "    except Exception as e:\n",
    "        print(f\"查询执行出错: {e}\")\n",
    "        return None\n",
    "\n",
    "# 你的自定义查询区域\n",
    "# 在这里编写和执行你自己的SQL查询\n",
    "\n",
    "# 示例：\n",
    "# my_query = \"SELECT COUNT(*) FROM your_table WHERE condition\"\n",
    "# result = await execute_custom_query(my_query)\n",
    "# print(result)\n",
    "\n",
    "print(\"自定义查询区域 - 请在上方编写你的查询\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 数据导出功能"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 数据导出功能\n",
    "async def export_data_to_csv(query, filename, params=None):\n",
    "    \"\"\"将查询结果导出为CSV文件\"\"\"\n",
    "    try:\n",
    "        df = await execute_custom_query(query, params)\n",
    "        if df is not None and not df.empty:\n",
    "            # 确保导出目录存在\n",
    "            export_dir = os.path.join(os.getcwd(), 'exports')\n",
    "            os.makedirs(export_dir, exist_ok=True)\n",
    "            \n",
    "            filepath = os.path.join(export_dir, filename)\n",
    "            df.to_csv(filepath, index=False, encoding='utf-8-sig')\n",
    "            print(f\"数据已导出到: {filepath}\")\n",
    "            print(f\"导出行数: {len(df)}\")\n",
    "            return filepath\n",
    "        else:\n",
    "            print(\"没有数据可导出\")\n",
    "            return None\n",
    "    except Exception as e:\n",
    "        print(f\"导出数据时出错: {e}\")\n",
    "        return None\n",
    "\n",
    "# 示例导出\n",
    "if target_table:\n",
    "    export_query = f\"SELECT * FROM {full_table_name} LIMIT 1000\"\n",
    "    timestamp = datetime.now().strftime(\"%Y%m%d_%H%M%S\")\n",
    "    export_filename = f\"{target_table}_sample_{timestamp}.csv\"\n",
    "    \n",
    "    print(f\"导出 {target_table} 的样本数据...\")\n",
    "    exported_file = await export_data_to_csv(export_query, export_filename)\n",
    "    \n",
    "    if exported_file:\n",
    "        print(f\"导出完成: {exported_file}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 清理和关闭连接"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# 关闭数据库连接\n",
    "try:\n",
    "    await db.close()\n",
    "    print(\"数据库连接已关闭\")\n",
    "except Exception as e:\n",
    "    print(f\"关闭连接时出错: {e}\")\n",
    "\n",
    "print(\"\\n=== 分析完成 ===\")\n",
    "print(\"本笔记本提供了完整的数据库交互式分析模板\")\n",
    "print(\"你可以根据需要修改查询和分析逻辑\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 总结\n",
    "\n",
    "本笔记本演示了如何使用 alphahome 框架进行数据库交互式分析，包括：\n",
    "\n",
    "1. **环境设置**: 导入必要的库和配置项目路径\n",
    "2. **数据库连接**: 使用 ConfigManager 和 DBManager 建立连接\n",
    "3. **结构探索**: 查看数据库中的表和列结构\n",
    "4. **数据查询**: 执行各种查询获取数据\n",
    "5. **质量检查**: 检查数据完整性和质量\n",
    "6. **自定义查询**: 提供灵活的查询接口\n",
    "7. **数据导出**: 将分析结果导出为CSV文件\n",
    "\n",
    "### 使用建议\n",
    "\n",
    "- 根据你的具体数据结构调整查询语句\n",
    "- 在处理大量数据时注意使用适当的LIMIT子句\n",
    "- 定期保存重要的分析结果\n",
    "- 可以将常用的分析逻辑封装成函数以便重复使用\n",
    "\n",
    "### 扩展功能\n",
    "\n",
    "你可以基于这个模板添加更多功能：\n",
    "- 机器学习模型训练\n",
    "- 更复杂的统计分析\n",
    "- 交互式图表\n",
    "- 自动化报告生成\n",
    "- 数据监控和告警"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}